package org.zhaya.sdk.executor;

import cn.hutool.core.util.ObjectUtil;
import com.google.common.base.Stopwatch;
import io.logz.sawmill.*;
import io.logz.sawmill.exceptions.PipelineExecutionException;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.zhaya.sdk.ProcessorInterceptor;
import org.zhaya.sdk.processors.TransposeProcessor;
import org.zhaya.sdk.swamill.ProcessDoc;

import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static java.util.concurrent.TimeUnit.NANOSECONDS;

@Slf4j
public class PipelineExecutor {

    private final PipelineExecutionTimeWatchdog watchdog;
    private final PipelineExecutionMetricsTracker pipelineExecutionMetricsTracker;
    private final List<ProcessorInterceptor> processorInterceptors;

    public PipelineExecutor() {
        this(new PipelineExecutionMetricsMBean());
    }

    public PipelineExecutor(PipelineExecutionMetricsTracker pipelineExecutionMetricsTracker) {
        this(new PipelineExecutionTimeWatchdog(100, 1000, pipelineExecutionMetricsTracker, context -> {
        }), pipelineExecutionMetricsTracker);
    }

    public PipelineExecutor(PipelineExecutionTimeWatchdog watchdog, PipelineExecutionMetricsTracker pipelineExecutionMetricsTracker) {
        this(watchdog, pipelineExecutionMetricsTracker, Collections.emptyList());
    }

    public PipelineExecutor(PipelineExecutionTimeWatchdog watchdog, PipelineExecutionMetricsTracker pipelineExecutionMetricsTracker, List<ProcessorInterceptor> processorInterceptors) {
        this.watchdog = watchdog;
        this.pipelineExecutionMetricsTracker = pipelineExecutionMetricsTracker;
        this.processorInterceptors = processorInterceptors;
    }

    public PipelineResult execute(Pipeline pipeline, ProcessDoc doc) {
        PipelineStopwatch pipelineStopwatch = new PipelineStopwatch().start();

        long executionIdentifier = watchdog.startedExecution(pipeline.getId(), doc, Thread.currentThread());

        PipelineResult pipelineResult;
        ExecutionResult executionResult;
        try {
            List<ExecutionStep> executionSteps = pipeline.getExecutionSteps();
            pipelineResult = executeSteps(executionSteps, pipeline, doc, pipelineStopwatch);
            executionResult = pipelineResult.getResult();

            // Prevent race condition with watchdog - check whether the execution got interrupted
            boolean hasBeenInterrupted = watchdog.stopWatchedPipeline(executionIdentifier);

            if (hasBeenInterrupted) {
                // clear interrupted flag
                Thread.interrupted();
                executionResult = ExecutionResult.expired(pipelineStopwatch.pipelineElapsed(MILLISECONDS));
            } else if (watchdog.isOvertime(executionIdentifier)) {
                executionResult = ExecutionResult.overtime(executionResult, pipelineStopwatch.pipelineElapsed(MILLISECONDS));
            }
        } catch (RuntimeException e) {
            pipelineExecutionMetricsTracker.pipelineFailedOnUnexpectedError(pipeline.getId(), doc, e);
            throw new PipelineExecutionException(pipeline.getId(), e);

        } finally {
            pipelineStopwatch.stop();
            watchdog.removeExecution(executionIdentifier);
        }

        if (executionResult.isSucceeded()) {
            log.trace("pipeline {} executed successfully, took {}ns", pipeline.getId(), pipelineStopwatch.pipelineElapsed());
            pipelineExecutionMetricsTracker.pipelineFinishedSuccessfully(pipeline.getId(), doc, pipelineStopwatch.pipelineElapsed());

        } else if (executionResult.isDropped()) {
            pipelineExecutionMetricsTracker.docDropped(pipeline.getId(), doc);
        } else {
            pipelineExecutionMetricsTracker.pipelineFailed(pipeline.getId(), doc);
        }

        return pipelineResult;
    }

    private PipelineResult executeSteps(List<ExecutionStep> executionSteps, Pipeline pipeline, ProcessDoc doc, PipelineStopwatch pipelineStopwatch) {
        Set<ProcessDoc> docSet = new HashSet<>();
        ExecutionResult executionResult = null;
        for (ExecutionStep executionStep : executionSteps) {
            executionResult = executeMultiStep(executionStep, pipeline, doc, pipelineStopwatch, docSet);
            if (Objects.nonNull(executionResult)) {
                boolean shouldStop = (pipeline.isStopOnFailure() && executionResult.isFailed()) || executionResult.isExpired() || executionResult.isDropped();
                if (shouldStop) {
                    break;
                }
            }
        }
        return new PipelineResult(ObjectUtil.defaultIfNull(executionResult, ExecutionResult.success()), docSet);
    }

    private ExecutionResult executeMultiStep(ExecutionStep executionStep, Pipeline pipeline, ProcessDoc doc, PipelineStopwatch pipelineStopwatch, Set<ProcessDoc> docSet) {
        ExecutionResult result = null;
        boolean multiFieldsName = doc.hasField(TransposeProcessor.MULTI_DOC_FIELD, false);
        if (multiFieldsName) {
            docSet.remove(doc);
            List<ProcessDoc> docs = doc.getField(TransposeProcessor.MULTI_DOC_FIELD, false);
            for (ProcessDoc document : docs) {
                result = executeSingleStep(executionStep, pipeline, document, pipelineStopwatch);
                docSet.add(document);
            }
        } else {
            result = executeSingleStep(executionStep, pipeline, doc, pipelineStopwatch);
            docSet.add(doc);
        }
        return result;
    }


    private ExecutionResult executeSingleStep(ExecutionStep executionStep, Pipeline pipeline, ProcessDoc doc, PipelineStopwatch pipelineStopwatch) {
        try {
            if (executionStep instanceof ConditionalExecutionStep) {
                return executeConditionalStep((ConditionalExecutionStep) executionStep, pipeline, doc, pipelineStopwatch);
            } else if (executionStep instanceof ProcessorExecutionStep) {
                return executeProcessorStep((ProcessorExecutionStep) executionStep, pipeline, doc, pipelineStopwatch);
            }
        } catch (InterruptedException e) {
            log.error("interrupted pipeline id: {}", pipeline.getId(), e);
            Thread.currentThread().interrupt();
            return ExecutionResult.expired();
        }
        throw new RuntimeException("Unsupported execution step " + executionStep.getClass());
    }

    private ExecutionResult executeConditionalStep(ConditionalExecutionStep conditionalExecutionStep, Pipeline pipeline, ProcessDoc doc, PipelineStopwatch pipelineStopwatch) throws InterruptedException {
        Condition condition = conditionalExecutionStep.getCondition();

        if (condition.evaluate(doc)) {
            return executeSteps(conditionalExecutionStep.getOnTrue(), pipeline, doc, pipelineStopwatch).getResult();
        } else {
            return executeSteps(conditionalExecutionStep.getOnFalse(), pipeline, doc, pipelineStopwatch).getResult();
        }
    }

    private ExecutionResult executeProcessorStep(ProcessorExecutionStep executionStep, Pipeline pipeline, ProcessDoc doc, PipelineStopwatch pipelineStopwatch) throws InterruptedException {
        Processor processor = executionStep.getProcessor();
        String pipelineId = pipeline.getId();
        String processorName = executionStep.getProcessorName();
        pipelineStopwatch.startProcessor();
        ProcessResult processResult = new ProcessProxy(processor, pipeline, processorInterceptors).process(doc);
        long processorTook = pipelineStopwatch.processorElapsed();

        if (processResult.isSucceeded()) {
            pipelineExecutionMetricsTracker.processorFinishedSuccessfully(pipelineId, processorName, processorTook);
            Optional<List<ExecutionStep>> onSuccessExecutionSteps = executionStep.getOnSuccessExecutionSteps();
            if (onSuccessExecutionSteps.isPresent()) {
                return executeSteps(onSuccessExecutionSteps.get(), pipeline, doc, pipelineStopwatch).getResult();
            }
            return ExecutionResult.success();
        } else if (processResult.isDropped()) {
            return ExecutionResult.dropped();
        } else {
            Optional<List<ExecutionStep>> onFailureExecutionSteps = executionStep.getOnFailureExecutionSteps();
            if (onFailureExecutionSteps.isPresent()) {
                return executeSteps(onFailureExecutionSteps.get(), pipeline, doc, pipelineStopwatch).getResult();
            } else {
                pipelineExecutionMetricsTracker.processorFailed(pipelineId, processorName, doc);
                return processorErrorExecutionResult(processResult.getError().get(), processorName, pipeline);
            }
        }
    }

    private ExecutionResult processorErrorExecutionResult(ProcessResult.Error error, String processorName, Pipeline pipeline) {
        String message = error.getMessage();
        if (error.getException().isPresent()) {
            return ExecutionResult.failure(message, processorName,
                    new PipelineExecutionException(pipeline.getId(), error.getException().get()));
        } else {
            return ExecutionResult.failure(message, processorName);
        }
    }

    @NoArgsConstructor
    private static class PipelineStopwatch {
        private Stopwatch stopwatch;
        private long processorStartElapsedTime;
        private final TimeUnit timeUnit = NANOSECONDS;

        public PipelineStopwatch start() {
            stopwatch = Stopwatch.createStarted();
            processorStartElapsedTime = 0;
            return this;
        }

        public long pipelineElapsed() {
            return stopwatch.elapsed(timeUnit);
        }

        public long pipelineElapsed(TimeUnit timeUnit) {
            return stopwatch.elapsed(timeUnit);
        }

        public long processorElapsed() {
            return stopwatch.elapsed(timeUnit) - processorStartElapsedTime;
        }

        public void startProcessor() {
            processorStartElapsedTime = stopwatch.elapsed(timeUnit);
        }

        public void stop() {
            stopwatch.stop();
        }
    }

    private static class ProcessProxy {

        private final Processor processor;
        private final Pipeline pipeline;
        private final List<ProcessorInterceptor> processorInterceptors;
        private final boolean isHandler;

        public ProcessProxy(Processor processor, Pipeline pipeline, List<ProcessorInterceptor> processInterceptors) {
            this.processor = processor;
            this.pipeline = pipeline;
            this.isHandler = CollectionUtils.isNotEmpty(processInterceptors);
            if (isHandler) {
                this.processorInterceptors = processInterceptors.stream().filter(ProcessorInterceptor::getState).sorted(Comparator.comparingInt(ProcessorInterceptor::getOrder)).collect(Collectors.toList());
            } else {
                this.processorInterceptors = Collections.emptyList();
            }
        }

        public ProcessResult process(ProcessDoc doc) {
            if (isHandler) {
                for (ProcessorInterceptor processorInterceptor : processorInterceptors) {
                    processorInterceptor.onBefore(processor, doc, pipeline);
                }
            }
            ProcessResult result = null;
            try {
                result = processor.process(doc);
                if (isHandler) {
                    for (ProcessorInterceptor processorInterceptor : processorInterceptors) {
                        processorInterceptor.onAfter(processor, doc, pipeline, result);
                    }
                }
            } catch (InterruptedException interruptedException) {
                if (isHandler) {
                    for (ProcessorInterceptor processorInterceptor : processorInterceptors) {
                        processorInterceptor.onThrowing(processor, doc, pipeline, interruptedException);
                    }
                }
            }
            return result;
        }
    }
}